package com.gin.monitor.agg.utils;

import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author gin
 * @date 2021/2/25
 */
public class JobEnvUtils {


    public static StreamExecutionEnvironment initEnv(String[] args) {
        final ParameterTool params = ParameterTool.fromArgs(args);
        // for cluster
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 默认并行度为1
        env.setParallelism(1);
        // 从 args 中获取参数
        env.getConfig().setGlobalJobParameters(params);

        // 设置checkpoint模式, 精确一次消费
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        // checkpoints 超时时间
        env.getCheckpointConfig().setCheckpointTimeout(5 * 60 * 1000);
        /*
         * 设置checkpoint任务之间的间隔时间  checkpoint job1  checkpoint job2
         * 防止触发太密集的flink checkpoint，导致消耗过多的flink集群资源
         * 导致影响整体性能
         * 1000ms
         * 注意: 设置了这个参数, 表示checkpoint就应该是串行执行的;
         * 即: setMaxConcurrentCheckpoints(1), 否则当前配置会失效
         */
        // 设置checkpoint最大并行的个数
        // 1 表示串行执行
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
        // flink 任务取消之后，checkpoint数据是否删除
        // RETAIN_ON_CANCELLATION 当任务取消，checkpoints数据会保留
        // DELETE_ON_CANCELLATION 当任务取消，checkpoints数据会删除
        env.getCheckpointConfig().enableExternalizedCheckpoints(
            CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

        return env;
    }

}
